package com.swak.reactivex.stream;

import java.util.concurrent.atomic.AtomicLong;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;

/**
 * 数据流
 * 
 * @author 618lf
 */
@Getter
@Setter
@Accessors(chain = true)
@ToString
public class Stream {

	private static final AtomicLong INVOKE_ID = new AtomicLong(1);

	/** 内存分配 */
	private static final ByteBufAllocator Allocator = ByteBufAllocator.DEFAULT;
	private static final int ARQ_HEAD = 20;

	/** message ID **/
	private long id;
	/** message分片segment的序号(编码时：int) **/
	private long sn;
	/*** 数据大小 **/
	private int length;

	/*** 数据 (扩展头部 + 载体) **/
	private ByteBuf body;

	public Stream() {
		this.id = INVOKE_ID.getAndIncrement();
	}

	public Stream(long id) {
		this.id = id;
	}

	public Stream(long id, long sn, int length) {
		this.id = id;
		this.sn = sn;
		this.length = length;
	}

	public int getTotalLength() {
		return length + ARQ_HEAD;
	}

	/**
	 * 序列化 Stream
	 * 
	 * @param segment
	 * @return
	 */
	public void encode(ByteBuf buffer) {
		buffer.writeLong(this.getId());
		buffer.writeLong(this.getSn());
		buffer.writeInt(this.getLength());
		ByteBuf data = this.getBody();
		if (data != null) {
			buffer.writeBytes(data, data.readerIndex(), data.readableBytes());
		}
	}

	/**
	 * 序列化 Stream
	 * 
	 * @param segment
	 * @return
	 */
	public static ByteBuf encode(Stream stream) {
		ByteBuf buffer = Allocator.ioBuffer();
		buffer.writeLong(stream.getId());
		buffer.writeLong(stream.getSn());
		buffer.writeInt(stream.getLength());
		ByteBuf data = stream.getBody();
		if (data != null) {
			buffer.writeBytes(data, data.readerIndex(), data.readableBytes());
		}

		// 返回实际的长度
		ByteBuf slice = buffer.readRetainedSlice(buffer.readableBytes());
		buffer.release();

		// 需要写入输出 buffer 后，才能释放
		return slice;
	}

	/**
	 * 解码 Stream
	 * 
	 * @param buffer
	 * @return
	 */
	public static Stream decode(ByteBuf buffer) {
		if (buffer.readableBytes() < ARQ_HEAD) {
			return null;
		}
		Stream stream = new Stream(buffer.readLong(), // id
				buffer.readLong(), // sn
				buffer.readInt()); // length

		// 返回数据 : 有数据则写入数据，否则不会写入数据
		int len = stream.getLength();
		return len > 0 ? stream.setBody(buffer.readRetainedSlice(len)) : stream;
	}
}
